import java.lang.*;
import java.util.*;
import java.io.IOException;

import org.apache.hadoop.fs.Path;
import org.apache.hadoop.io.IntWritable;
import org.apache.hadoop.io.Text;
import org.apache.hadoop.io.LongWritable;
import org.apache.hadoop.conf.Configuration;

import org.apache.hadoop.mapreduce.Job;
import org.apache.hadoop.mapreduce.Mapper;
import org.apache.hadoop.mapreduce.Reducer;
import org.apache.hadoop.mapreduce.Mapper.Context;
import org.apache.hadoop.mapreduce.lib.input.FileInputFormat;
import org.apache.hadoop.mapreduce.lib.output.FileOutputFormat;

public class MapReduceSample {

	static class MapClass extends Mapper<LongWritable, Text, Text, IntWritable> {
		
		public void map(LongWritable key, Text value, Context context) 
				throws IOException, InterruptedException {
			
			String line = value.toString();
			String[] items = line.split(" ");
			
			// output mapreduce
			for(int i=0; i<items.length; i++) {
				context.write(new Text(items[i]), new IntWritable(1));
			}
		}
	}
	
	static class ReduceClass extends Reducer<Text, IntWritable, Text, IntWritable> {
		
		public void reduce(Text key, Iterable<IntWritable> values, Context context) 
				throws IOException, InterruptedException {
		
				int num = 0;
				
				for (IntWritable value : values) {
					num += value.get();
				}
				
				context.write(key, new IntWritable(num));
			}
	}
	
	public static void main(String[] args) throws Exception{
		// TODO Auto-generated method stub
		
		/*if (args.length != 2) {
			System.err.println("Usage: MapReduceSample <input path> <output path>");
			System.exit(-1);
		}*/
		
		Configuration conf = new Configuration();
		Job job = new Job(conf, "wts_mapreduce_sample");
		
		job.setJarByClass(MapReduceSample.class);
		
		FileInputFormat.addInputPath(job, new Path(args[0]));
		FileOutputFormat.setOutputPath(job, new Path(args[1]));
		
		job.setMapperClass(MapClass.class);
		job.setReducerClass(ReduceClass.class);
		
		job.setOutputKeyClass(Text.class);
		job.setOutputValueClass(IntWritable.class);
		
		System.exit(job.waitForCompletion(true)? 0 : 1);
	}

}
